#include "Spark/Net/TcpConnection.h"
#include "Spark/Utils/Logger.h"
#include "Spark/Net/Channel.h"
#include "Spark/Net/EventLoop.h"
#include "Spark/Net/Socket.h"
#include <unistd.h>
#include <errno.h>

static constexpr size_t kHighWaterMarkLimit = 64 * 1024 * 1024;

static const char *stateToString(TcpConnection::TcpState state)
{
    switch (state)
    {
        case TcpConnection::TcpState::Disconnected:
            return "Disconnected";
        case TcpConnection::TcpState::Connecting:
            return "Connecting";
        case TcpConnection::TcpState::Connected:
            return "Connected";
        case TcpConnection::TcpState::Disconnecting:
            return "Disconnecting";
    }
    return "";
}

using LG_ = std::lock_guard<std::mutex>;

TcpConnection::TcpConnection(EventLoop *loop,
                             int fd,
                             std::string name,
                             const SocketAddr &local,
                             const SocketAddr &peer)
    : loop_(loop),
      socket_(new Socket(fd)),
      channel_(new Channel(loop_, fd)),
      name_(std::move(name)),
      local_(local),
      peer_(peer),
      state_(TcpState::Connecting),
      highWatermark_(kHighWaterMarkLimit)
{
    channel_->setCallback(CBType::READ, [this] {
        TimePoint now = TimePoint::Now();
        handleRead(now);
    });
    channel_->setCallback(CBType::WRITE, [this] { handleWrite(); });
    channel_->setCallback(CBType::CLOSE, [this] { handleClose(); });
    channel_->setCallback(CBType::ERROR, [this] { handleError(); });
    LOG_DEBUG << "TcpConnection created, fd:" << fd;
    socket_->setSocketOption(SO_::KEEPALIVE, 1);
}

TcpConnection::~TcpConnection()
{
    LOG_DEBUG << "TcpConnection destroyed, fd:" << socket_->fd()
              << " state: " << stateToString(state_);
    assert(state_ == TcpState::Disconnected);
}

bool TcpConnection::getTcpInfo(struct tcp_info *tcpi) const
{
    return socket_->getTcpInfo(tcpi);
}

std::string TcpConnection::getTcpInfoString() const
{
    char buf[1024]{};
    socket_->getTcpInfoString(buf, sizeof(buf));
    return buf;
}

void TcpConnection::send(MsgBuffer *buf)
{
    if (state_ == TcpState::Connected)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(buf->peek(), buf->readableBytes());
            buf->retrieveAll();
        }
        else
        {
            auto thisPtr = shared_from_this();
            loop_->runInLoop([thisPtr, buf] {
                thisPtr->sendInLoop(buf->peek(), buf->readableBytes());
            });
        }
    }
}

void TcpConnection::send(std::string buf)
{
    if (state_ == TcpState::Connected)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(buf.data(), buf.size());
        }
        else
        {
            auto thisPtr = shared_from_this();
            loop_->runInLoop([thisPtr, buf] {
                thisPtr->sendInLoop(buf.data(), buf.size());
            });
        }
    }
}

void TcpConnection::sendInLoop(const void *data, size_t len)
{
    assert(loop_->isInLoopThread());
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;
    if (state_ == TcpState::Disconnected)
    {
        LOG_WARN << "disconnected, give up writing";
        return;
    }
    // if no thing in output queue, try writing directly
    if (!channel_->isWriteEvent() && outputBuffer_.readableBytes() == 0)
    {
        nwrote = ::write(channel_->fd(), data, len);
        if (nwrote >= 0)
        {
            remaining = len - nwrote;
            if (remaining == 0 && writeDoneCallback_)
            {
                loop_->queueInLoop(
                    std::bind(writeDoneCallback_, shared_from_this()));
            }
        }
        else  // nwrote < 0
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK)
            {
                LOG_SYSERR << "TcpConnection::sendInLoop";
                if (errno == EPIPE ||
                    errno == ECONNRESET)  // FIXME: any others?
                {
                    faultError = true;
                }
            }
        }
    }
    assert(remaining <= len);
    if (!faultError && remaining > 0)
    {
        size_t oldLen = outputBuffer_.readableBytes();
        if (oldLen + remaining >= highWatermark_ && oldLen < highWatermark_ &&
            highWatermarkCallback_)
        {
            loop_->queueInLoop(std::bind(highWatermarkCallback_,
                                         shared_from_this(),
                                         oldLen + remaining));
        }
        outputBuffer_.append(static_cast<const char *>(data) + nwrote,
                             remaining);
        if (!channel_->isWriteEvent())
        {
            channel_->addWriteEvent();
        }
    }
}

void TcpConnection::setTcpNoDelay(socklen_t optval)
{
    socket_->setTCPNoDelay(optval);
}

void TcpConnection::connectEstablished()
{
    assert(state_ == TcpState::Connecting);
    state_ = TcpState::Connected;
    channel_->tieCallbackOwner(shared_from_this());
    channel_->addReadEvent();
    connectionCallback_(shared_from_this());
}

void TcpConnection::connectDestroyed()
{
    assert(loop_->isInLoopThread());
    if (state_ == TcpState::Connected)
    {
        state_ = TcpState::Disconnected;
        channel_->disableAll();
        connectionCloseCallback_(shared_from_this());
    }
    channel_->remove();
}

void TcpConnection::forceClose()
{
    if (state_ == TcpState::Connected || state_ == TcpState::Disconnecting)
    {
        state_ = TcpState::Disconnecting;
        loop_->queueInLoop(
            std::bind(&TcpConnection::handleClose, shared_from_this()));
    }
}

void TcpConnection::shutdown()
{
    if (state_ == TcpState::Connected)
    {
        state_ = TcpState::Disconnecting;
        loop_->runInLoop([this] { shutdownInLoop(); });
    }
}

void TcpConnection::startRead()
{
    loop_->runInLoop([this]{ startReadInLoop(); });
}

void TcpConnection::stopRead()
{
    loop_->runInLoop([this]{ stopReadInLoop(); });
}

void TcpConnection::shutdownInLoop()
{
    assert(loop_->isInLoopThread());
    if (!channel_->isWriteEvent())
    {
        socket_->closeWrite();
    }
}

void TcpConnection::startReadInLoop()
{
    assert(loop_->isInLoopThread());
    if (!reading_ || !channel_->isReadEvent())
    {
        channel_->addReadEvent();
        reading_ = true;
    }
}

void TcpConnection::stopReadInLoop()
{
    assert(loop_->isInLoopThread());
    if (reading_ || !channel_->isReadEvent())
    {
        channel_->removeReadEvent();
        reading_ = false;
    }
}

void TcpConnection::handleRead(TimePoint receiveTime)
{
    assert(loop_->isInLoopThread());
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if (n > 0)
    {
        messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
    }
    else if (n == 0)
    {
        handleClose();
    }
    else
    {
        errno = savedErrno;
        LOG_SYSERR << "TcpConnection::handleRead";
        handleError();
    }
}

void TcpConnection::handleWrite()
{
    assert(loop_->isInLoopThread());
    if (channel_->isWriteEvent())
    {
        ssize_t n = ::write(channel_->fd(),
                            outputBuffer_.peek(),
                            outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->removeWriteEvent();
                if (writeDoneCallback_)
                {
                    loop_->queueInLoop(
                        std::bind(writeDoneCallback_, shared_from_this()));
                }
                if (state_ == TcpState::Disconnecting)
                {
                    socket_->closeWrite();
                }
            }
        }
        else
        {
            LOG_SYSERR << "TcpConnection::handleWrite";
        }
    }
    else
    {
        LOG_TRACE << "Connection fd = " << channel_->fd()
                  << " is down, no more writing";
    }
}

void TcpConnection::handleClose()
{
    assert(loop_->isInLoopThread());
    assert(state_ == TcpState::Connected || state_ == TcpState::Disconnecting);
    LOG_TRACE << "fd = " << channel_->fd()
              << " state = " << stateToString(state_);
    state_ = TcpState::Disconnected;
    channel_->disableAll();
    auto thisPtr(shared_from_this());
    connectionCallback_(thisPtr);
    connectionCloseCallback_(thisPtr);
}

void TcpConnection::handleError()
{
    int err = Socket::getSockErrno(channel_->fd());
    LOG_ERROR << "TcpConnection::handleError [fd" << channel_->fd()
              << "] - SO_ERROR = " << strerror_tl(err);
}

void defaultConnectionCallback(const TcpConnectionPtr &p)
{
    LOG_TRACE << p->localAddr().toStringPort() << " -> "
              << p->peerAddr().toStringPort() << " is "
              << (p.get() ? "UP" : "DOWN");
}

void defaultMessageCallback(const TcpConnectionPtr &p,
                            MsgBuffer *buffer,
                            TimePoint)
{
    LOG_TRACE << "Message Received from:[" << p->peerAddr().toStringPort()
              << "] saying: ["
              << std::string(buffer->peek(), buffer->readableBytes()) << "]";
    buffer->retrieveAll();
}